package com.example.mqdemo.rocket.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Component
public class RocketListenerConfig {

    @Service
    @RocketMQMessageListener(consumerGroup = "myGroup", topic = "orderCreate")
    public class OrderCreateListener implements RocketMQListener<String> {

        @Override
        public void onMessage(String message) {
            log.info("rocket>>>>>>消费消息:{}", message);
        }

//        @Override
//        public void prepareStart(DefaultMQPushConsumer consumer) {
//            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
//            consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis()));
//        }
    }

    @RocketMQTransactionListener
    public class OrderCancelListener implements RocketMQLocalTransactionListener {

        private Map<String, Object> localTrans = new ConcurrentHashMap<>();

        /**
         * 当发送事务半消息成功时，将调用此方法执行本地事务
         *
         * @param msg 半消息
         * @param arg 自定义业务参数
         * @return
         */
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            try {

                log.info("【本地业务执行完毕】 msg:{}, arg:{}", msg, arg);

                localTrans.put(msg.getHeaders().get("orderId") + "", msg.getPayload());

                int i = 1 / 0;

                // 返回COMMIT之后，调用消费者(Listener)方法
                return RocketMQLocalTransactionState.COMMIT;

            } catch (Exception e) {

                localTrans.remove(msg.getHeaders().get("orderId") + "");

                log.error("【执行本地业务异常】 exception message:{}", e.getMessage());

                // 这里返回 UNKNOWN 的话，会创建定时任务调用checkLocalTransaction方法。
                // 如果返回 ROLLBACK 状态，则不会调用checkLocalTransaction方法和消费方法
                return RocketMQLocalTransactionState.UNKNOWN;

            }
        }

        /**
         * 当半消息未得到响应，broker将发送检查事务状态的消息，该方法就会被调用来获取本地事务的状态
         *
         * @param msg check message
         * @return
         */
        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
            log.info("【执行检查任务】");
            String orderId = msg.getHeaders().get("orderId") + "";
            boolean b = localTrans.containsKey(orderId);
            if (b) {
                // 返回COMMIT之后,停止调用该方法,会去调用消费者(Listener)方法
                return RocketMQLocalTransactionState.COMMIT;
            }
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }


}
